/*
 * 文件名：KafkaMessageConverterImpl.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：napo
 * 修改时间：2015-1-5
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.realmq;

import java.nio.charset.Charset;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.model.RealStreamDataPack;
import com.ailk.bigdata.etl.realstream.server.model.ValueEvent;
import com.ailk.bigdata.etl.realstream.server.monitor.EventCounter;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;
import com.lmax.disruptor.RingBuffer;

import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

/**
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2015-1-5]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class KafkaMessageConverter implements Runnable
{
	private static Logger logger = LoggerFactory.getLogger(KafkaMessageConverter.class);
	private EventCounter eventCounter;
	private int dataType;
	private String countMark;
	private final Charset charset = Charset.forName("utf-8");
	private KafkaStream<byte[], byte[]> mStream;
	private int threadNumber;

	public KafkaMessageConverter(EventCounter eventCounter, int dataType, KafkaStream<byte[], byte[]> mStream, int threadNumber)
	{
		this.eventCounter = eventCounter;
		this.dataType = dataType;
		this.countMark = Integer.toString(dataType);
		this.mStream = mStream;
		this.threadNumber = threadNumber;
	}

	public String convert(Object message) throws Exception
	{
		String received = new String((byte[]) message, charset);
		return received;
	}

	@Override
	public void run()
	{
		int reveivedCount = 0;
		ConsumerIterator<byte[], byte[]> ite = mStream.iterator();
		logger.info("Thread start: " + threadNumber);
		RealStreamDataPack rStreamDataPack = null;
		while (!Thread.interrupted() && ite.hasNext())
		{
			try
			{
				byte[] dataContent = ite.next().message();
				rStreamDataPack = RealStreamDataPack.build(dataType, dataContent);
				publicMessage(rStreamDataPack);
				logger.debug("Thread " + threadNumber + ": " + convert(dataContent));
				eventCounter.incrementEventReceivedByType(countMark);
				reveivedCount++;
			}
			catch (Exception e)
			{
				logger.error("kafka message accept exception: ", e);
			}
			if (reveivedCount % 10000 == 0)
			{
				logger.debug("线程[" + threadNumber + "]接收到消息[" + reveivedCount + "]");
			}
		}
		KafKaMessageConsumer.KAFKA_STREAM_CHACHE.put(threadNumber, mStream);
		logger.info("Shutting down Thread: " + threadNumber);
	}

	protected void publicMessage(RealStreamDataPack dPack) throws Exception
	{
		if (RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.containsKey(dPack.getDataType()))
		{
			RingBuffer<ValueEvent> ringBuffer = RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.get(dPack.getDataType());
			long next = ringBuffer.next();
			try
			{

				ValueEvent event = ringBuffer.get(next);
				event.setDataPackage(dPack);
			}
			finally
			{
				ringBuffer.publish(next);
			}
		}
		else
		{
			logger.debug("接口单元[" + dPack.getDataType() + "]未注册 不能进行处理........");
		}
	}

	public KafkaStream<byte[], byte[]> getmStream()
	{
		return this.mStream;
	}

	public void setmStream(KafkaStream<byte[], byte[]> mStream)
	{
		this.mStream = mStream;
	}

	public int getThreadNumber()
	{
		return this.threadNumber;
	}

	public void setThreadNumber(int threadNumber)
	{
		this.threadNumber = threadNumber;
	}

}
